package com.qingyunge.app.func;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.qingyunge.util.DruidDSUtil;
import com.qingyunge.util.PhoenixUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.SQLException;

public class DimSinkFunction extends RichSinkFunction<JSONObject> {
    private DruidDataSource druidDataSource = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidDSUtil.createDataSource();
    }

    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        // 获取连接
        DruidPooledConnection connection = druidDataSource.getConnection();
        // 写出数据 {"sinkTable":"dim_table_test","database":"gmall","xid":12647,"data":{"name":"w53tqw3434w","id":6},"commit":true,"type":"insert","table":"table_test","ts":1718156399}
        String sinkTable = value.getString("sinkTable");
        JSONObject data = value.getJSONObject("data");
        try {
            PhoenixUtil.upsertValues(connection,sinkTable,data);
        } catch (SQLException throwables) {
            throwables.printStackTrace();
        }
        // 归还连接
        connection.close();
    }
}
